//Team 4 - Gold Coast
//Duy Kha Nguyen 598240 
//Rongkai Zhou 562946 
//Zhangyue Cheng 548692
//Binzan Liang 591911
//Wei Liu 572322
package couchdb;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;

import org.apache.log4j.Logger;
import org.lightcouch.CouchDbClient;
import org.lightcouch.Response;

public class ImportTweetTask implements Runnable {

    private static final Logger logger = Logger.getLogger(ImportTweetTask.class);

    private final CouchDbClient dataSource;

    private final BlockingQueue<Map> inputQueue;

    private boolean stop;

    public ImportTweetTask(CouchDbClient dataSource, BlockingQueue<Map> inputQueue) {
        super();
        this.dataSource = dataSource;
        this.inputQueue = inputQueue;
        this.stop = false;
    }

    public void stop() {
        this.stop = true;
    }

    public void run() {
        try {
            final List<Map> tweets = new ArrayList<Map>(1024);
            while (!stop || inputQueue.size() > 0) {
                inputQueue.drainTo(tweets, 1024);
                if (tweets.size() > 0) {
                    final List<Response> responses = dataSource.bulk(tweets, false);
                    for (final Response response : responses) {
                        if (response.getError() != null && !response.getError().isEmpty()) {
                            logger.info(response.getError() + ": " + response.getReason());
                        }
                    }
                    tweets.clear();
                } else {
                    try {
                        logger.info("sleeping");
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        } catch (Exception e) {
            logger.error("Unexpected exception", e);
        }
    }

}
